package com.zhao.biz.tag.perform

import ch.hsr.geohash.GeoHash
import com.zhao.utils.BaiduLBSHandler
import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.Row
import redis.clients.jedis.JedisCluster

/**
 * Description: 商圈标签<br/>
 * Copyright (c) ，2021 ， 赵 <br/>
 * A wet person does not fear the rain. <br/>
 * Date： 2021/1/14 14:44
 *
 * @author 柒柒
 * @version : 1.0
 */

object TagsBusiness extends IMakeTags {
  /**
   * 数据标签接口
   *
   * @param args
   * @return
   */
  override def makeTags(args: Any*): List[(String, Int)] = {

    val row = args(0).asInstanceOf[Row]
    val jedisCluster = args(1).asInstanceOf[JedisCluster]

    var list: List[(String, Int)] = List[(String, Int)]()

    //经度
    val longitudeStr = row.getAs[String]("_long")

    //维度
    val latitudeStr = row.getAs[String]("lat")

    //中国的经纬度范围大约是:经度73.66~135.05,维度3.86~53.55,不在范围内的数据不需要处理
    if (StringUtils.isNoneBlank(longitudeStr) && StringUtils.isNoneBlank(latitudeStr)) {
      val longitude = longitudeStr.toDouble
      val latitude = latitudeStr.toDouble
      if ((longitude >= 73.66 && longitude <=135.05) && (latitude >=3.86 && latitude <= 53.55)) {
        val geoHashStr = GeoHash.geoHashStringWithCharacterPrecision(latitude, longitude, 8)
        //返回的是地标(商圈)信息(前提:将日志文件中每天日志信息中的维度通过GeoHash算法已经转换为商圈信息,已经存储到redis分布式集群中了)
        var business = jedisCluster.hget("geoHashs", geoHashStr)

        //没有
        if (StringUtils.isBlank(business)) {
          //查询百度地图,得到经纬度对应的商圈信息,再存入redis server中
          business = BaiduLBSHandler.parseBusinessTagBy(longitudeStr, latitudeStr)
          if (StringUtils.isNoneBlank(business)) {
            jedisCluster.hset("geoHashs",geoHashStr,business)
          }

          //business有两种可能性: 可能性1->redisServer;可能性2->百度
          if (StringUtils.isNoneBlank(business)) {
            business.split(",").foreach(t => list :+= (t,1))
          }
        }
      }
    }

    list
  }
}




































